package com.example.zz.example.pattern.rxjava;

import android.app.ProgressDialog;
import android.content.Context;
import android.util.Log;
import android.widget.Toast;

import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Action;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

/**
 * Describe:
 *
 * @author zhangzhuo18.
 * Date: 2023/8/16
 */
public class RxJavaExample {
    public static final String TAG = "RxJavaExample";
    private ProgressDialog progressDialog;

    /**
     * 当这段代码执行时，首先会弹出一个处理对话框，然后在后台线程上执行complexCalculation。
     * 计算完成后，在另一个后台线程上执行阶乘计算。最后，结果字符串会在主线程上生成并显示为一个Toast消息，同时处理对话框会被隐藏。
     *
     * @param context
     */
    public void startExample(Context context) {
        Observable.defer(() -> Observable.just(complexCalculation(1, 1))
                        .delay(3, TimeUnit.SECONDS))
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSubscribe(disposable -> showProcessDialog(context))
                .doFinally(this::hideProcessDialog)
                .flatMap(result -> Observable.defer(() -> Observable.just(result)
                        .observeOn(Schedulers.io())
                        .map(this::calculateFactorial)
                        .observeOn(AndroidSchedulers.mainThread())
                        .map(factorial -> "Result: " + result + "\nFactorial Result: " + factorial)
                ))
                .subscribe(s -> showToast(s, context));

    }

    /**
     * 不使用lambda表达式，使用内部类写法
     *
     * @param context
     */
    public void startExample2(Context context) {

        Observable.defer(new Callable<ObservableSource<Integer>>() {
                    @Override
                    public ObservableSource<Integer> call() throws Exception {
                        return Observable.just(complexCalculation(1, 1));
                    }
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        showProcessDialog(context);
                    }
                })
                .doFinally(new Action() {
                    @Override
                    public void run() throws Exception {
                        hideProcessDialog();
                    }
                })
                .flatMap(new Function<Integer, ObservableSource<String>>() {
                    @Override
                    public ObservableSource<String> apply(Integer result) throws Exception {
                        return Observable.defer(new Callable<ObservableSource<? extends String>>() {
                            @Override
                            public ObservableSource<? extends String> call() throws Exception {
                                return Observable.just(result)
                                        .observeOn(Schedulers.io())
                                        .map(new Function<Integer, Integer>() {
                                            @Override
                                            public Integer apply(Integer integer) throws Exception {
                                                return calculateFactorial(integer);
                                            }
                                        }).observeOn(AndroidSchedulers.mainThread())
                                        .map(new Function<Integer, String>() {
                                            @Override
                                            public String apply(Integer factorial) throws Exception {
                                                return "Result: " + result + "\nFactorial Result: " + factorial;
                                            }
                                        });
                            }
                        });
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        showToast(s, context);
                    }
                });

    }


    /**
     * 操作符执行顺序
     * <p>
     * 在单一线程情况下，rxjava各类回调及操作符执行顺序，按下面分的模块和顺序进行回调，所有doOn开头的，都只推荐写一个，doOnNext可以写多个
     * doOnSubscribe -> onSubscribe -> subscribe -> doOnNext(map/flatMap/...) -> onNext -> doOnComplete -> onComplete -> doFinally
     * <p>
     * 如下面案例，添加了2个doOnSubscribe(这种写法不推荐)，3个map，2个doOnNext，2个doOnComplete，2个doFinally，并且排布顺序错乱，他们的调用顺序就满足上面规则
     * 首先调用doOnSubscribe模块，看代码在不同位置写了2个doOnSubscribe其中doOnSubscribe2在doOnSubscribe上面，直接运行时，doOnSubscribe2在doOnSubscribe之前被调用，然而，当在2个doOnSubscribe之间添加subscribeOn进行切换线程时，doOnSubscribe又在doOnSubscribe2之前被调用，所以这里只推荐写一个
     * 然后onSubscribe被调用
     * 然后subscribe被调用
     * 然后开始发射数据，发射数据后要先执行doOnNext(map/flatMap/...)模块，即使代码中doFinally、doOnComplete写在了doOnNext前面也不会执行，模块中包含的按顺序执行
     * 所以日志打印顺序  map -> doOnNext -> map1 -> doOnNext2 -> map2
     * 然后onNext被调用
     * 然后doOnComplete被调用，多个时从上往下调用，doOnComplete -> doOnComplete2
     * 然后doFinally被调用，doFinally2却被先调用，和doOnComplete相反，所以doOn开头的，除了doOnNext，其他不要多写！
     * <p>
     * <p>
     * <p>
     * <p>
     * doOn操作符
     * doOn操作符，不会改变数据流，主要用来监听数据流逻辑和起一些副作用！
     * <p>
     * doOnSubscribe：当Observable开始发射数据之前（即订阅发生时），执行指定的操作。这可以用于记录订阅开始的时间或执行一些初始化工作。
     * doOnNext：当Observable发射数据时，即调用onNext()时，执行指定的操作。这允许你在每个数据项被处理之前或之后执行某些操作，例如记录日志或更新UI。
     * doOnError：当Observable发生错误时，执行指定的操作。这对于在错误发生时执行清理工作或记录错误信息非常有用。
     * doOnComplete：当Observable正常完成时，即调用onComplete()时，执行指定的操作。这对于在流结束时执行一些收尾工作很有用，比如关闭资源或发送完成通知。
     * doOnDispose：当Observable的订阅被取消时，执行指定的操作。这对于在取消订阅时执行清理工作或记录信息很有用。
     * doFinally：无论Observable是正常完成、发生错误还是被取消订阅，都会执行指定的操作。这可以用于确保在Observable的生命周期结束时执行一些必要的操作。
     * <p>
     * <p>
     * 2024-04-17 10:04:25.140  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder :main
     * 2024-04-17 10:04:25.150  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder doOnSubscribe2:main
     * 2024-04-17 10:04:25.150  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder doOnSubscribe:main
     * 2024-04-17 10:04:25.150  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder onSubscribe:main
     * 2024-04-17 10:04:25.150  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder subscribe:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder map:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder doOnNext:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder map1:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder doOnNext2:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder map2:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder onNext:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder doOnComplete:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder doOnComplete2:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder onComplete:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder doFinally2:main
     * 2024-04-17 10:04:25.151  2772-2772  RxJavaExample           com.example.zz.example.check         I  testDoOnOrder doFinally:main
     */

    public void testDoOnOrder() {
        Log.i(TAG, "testDoOnOrder :" + Thread.currentThread().getName());
        Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
                        Log.i(TAG, "testDoOnOrder subscribe:" + Thread.currentThread().getName());
                        emitter.onNext("");
                        emitter.onComplete();
                        return;
                    }
                })
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        Log.i(TAG, "testDoOnOrder map:" + Thread.currentThread().getName());
                        return s;
                    }
                })
                .doFinally(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.i(TAG, "testDoOnOrder doFinally:" + Thread.currentThread().getName());
                    }
                })
                .doOnComplete(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.i(TAG, "testDoOnOrder doOnComplete:" + Thread.currentThread().getName());
                    }
                })
                .doOnNext(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.i(TAG, "testDoOnOrder doOnNext:" + Thread.currentThread().getName());
                    }
                })
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.i(TAG, "testDoOnOrder doOnSubscribe2:" + Thread.currentThread().getName());
                    }
                })
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        Log.i(TAG, "testDoOnOrder map1:" + Thread.currentThread().getName());
                        return s;
                    }
                })
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.i(TAG, "testDoOnOrder doOnSubscribe:" + Thread.currentThread().getName());
                    }
                })
                .doOnNext(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.i(TAG, "testDoOnOrder doOnNext2:" + Thread.currentThread().getName());
                    }
                })
                .doOnComplete(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.i(TAG, "testDoOnOrder doOnComplete2:" + Thread.currentThread().getName());
                    }
                })
                .doFinally(new Action() {
                    @Override
                    public void run() throws Exception {
                        Log.i(TAG, "testDoOnOrder doFinally2:" + Thread.currentThread().getName());
                    }
                })
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Exception {
                        Log.i(TAG, "testDoOnOrder map2:" + Thread.currentThread().getName());
                        return s;
                    }
                })
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "testDoOnOrder onSubscribe:" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(String s) {
                        Log.i(TAG, "testDoOnOrder onNext:" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "testDoOnOrder onError:" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "testDoOnOrder onComplete:" + Thread.currentThread().getName());
                    }
                });
    }

    /**
     * 线程切换的操作及回调打印案例
     * 默认情况下，Observable和Observer处于同一线程中，如果想要切换线程，则可以使用subscribeOn()和observeOn()
     * subscribeOn()影响向上流的线程调度，observeOn()影响向下流的线程调度
     * <p>
     * 1、以subscribe订阅为开始，开始运行后，首先回调Observer的onSubscribe通知用户，开始流的运行
     * 2、流向上运转，向上的过程只受到subscribeOn的控制进行切换线程，只会回调doOnSubscribe。
     * 多次切换线程多次添加doOnSubscribe进行回调监听都满足，下面案例subscribe之后向上首先切换为single线程，此时doOnSubscribe监听，打印显示doOnSubscribe:RxSingleScheduler-1运行在singel线程，继续向上流动
     * 其他操作符都不会产生影响，直到出现subscribeOn(Schedulers.io()将线程切换到io线程，后面又有doOnSubscribe监听，会被回调，此时打印在io线程RxCachedThreadScheduler-1
     * 3、继续向上流转，其他操作符依然不受影响，直到遇到subscribeOn(Schedulers.newThread()再次切换线程，继续向上流转到达向上流转的最高点ObservableOnSubscribe的subscribe，此时subscribe打印RxNewThreadScheduler-1，并调用onNext发送数据，调用onComplete()发送完成
     * 4、后面流向下流转，subscribeOn将不起作用，由observeOn进行线程切换，首先observeOn(Schedulers.single())将流切换到single，map1收到onNext发送的数据，打印为RxSingleScheduler-1；
     * 5、
     * <p>
     * 参考：https://zhuanlan.zhihu.com/p/68251074
     * <p>
     * 执行日志
     * 2024-03-15 17:03:11.171  3354-3354  RxJavaExample           com.example.zz.example.check         I  subscribe onSubscribe:main
     * 2024-03-15 17:03:11.173  3354-4703  RxJavaExample           com.example.zz.example.check         I  doOnSubscribe:RxSingleScheduler-1
     * 2024-03-15 17:03:11.190  3354-4704  RxJavaExample           com.example.zz.example.check         I  doOnSubscribe2:RxCachedThreadScheduler-1
     * 2024-03-15 17:03:11.191  3354-4705  RxJavaExample           com.example.zz.example.check         I  create:RxNewThreadScheduler-1
     * 2024-03-15 17:03:11.192  3354-4703  RxJavaExample           com.example.zz.example.check         I  map1 :RxSingleScheduler-1
     * 2024-03-15 17:03:11.192  3354-4703  RxJavaExample           com.example.zz.example.check         I  doOnComplete :RxSingleScheduler-1
     * 2024-03-15 17:03:11.194  3354-4706  RxJavaExample           com.example.zz.example.check         I  map2 :RxComputationThreadPool-1
     * 2024-03-15 17:03:11.194  3354-4706  RxJavaExample           com.example.zz.example.check         I  doOnComplete2 :RxComputationThreadPool-1
     * 2024-03-15 17:03:11.194  3354-4706  RxJavaExample           com.example.zz.example.check         I  doFinally :RxComputationThreadPool-1
     * 2024-03-15 17:03:11.194  3354-4707  RxJavaExample           com.example.zz.example.check         I  map3:RxNewThreadScheduler-2
     * 2024-03-15 17:03:11.194  3354-4707  RxJavaExample           com.example.zz.example.check         I  doOnNext1:RxNewThreadScheduler-2
     * 2024-03-15 17:03:11.195  3354-4708  RxJavaExample           com.example.zz.example.check         I  map4:RxCachedThreadScheduler-2
     * 2024-03-15 17:03:11.195  3354-4708  RxJavaExample           com.example.zz.example.check         I  doOnNext2:RxCachedThreadScheduler-2
     * 2024-03-15 17:03:11.195  3354-3354  RxJavaExample           com.example.zz.example.check         I  map5:main
     * 2024-03-15 17:03:11.195  3354-3354  RxJavaExample           com.example.zz.example.check         I  subscribe onNext:main
     * 2024-03-15 17:03:11.195  3354-3354  RxJavaExample           com.example.zz.example.check         I  subscribe onComplete:main
     */
    public void threadChangeExample() {

        Observable.create(new ObservableOnSubscribe<Integer>() {
                    @Override
                    public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                        Log.i(TAG, "create:" + Thread.currentThread().getName());
                        emitter.onNext(1);
                    }
                })
                .subscribeOn(Schedulers.newThread())
                .observeOn(Schedulers.single())
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        Log.i(TAG, "map1 :" + Thread.currentThread().getName());
                        return integer;
                    }
                })
                .observeOn(Schedulers.computation())
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        Log.i(TAG, "map2 :" + Thread.currentThread().getName());
                        return integer;
                    }
                })
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.i(TAG, "doOnSubscribe2:" + Thread.currentThread().getName());
                    }
                })
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        Log.i(TAG, "map3:" + Thread.currentThread().getName());
                        return integer;
                    }
                })
                .observeOn(AndroidSchedulers.mainThread())
                .doOnSubscribe(new Consumer<Disposable>() {
                    @Override
                    public void accept(Disposable disposable) throws Exception {
                        Log.i(TAG, "doOnSubscribe:" + Thread.currentThread().getName());
                    }
                })
                .subscribeOn(Schedulers.single())
                .map(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer) throws Exception {
                        Log.i(TAG, "map4:" + Thread.currentThread().getName());
                        return integer;
                    }
                }).subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.i(TAG, "subscribe onSubscribe:" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "subscribe onNext:" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "subscribe onError:" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "subscribe onComplete:" + Thread.currentThread().getName());
                    }
                });


    }

    private int complexCalculation(int a, int b) {
        // 复杂计算，用延时模拟
        return a * b;
    }

    private void showProcessDialog(Context context) {
        Log.i(TAG, "showProcessDialog:" + Thread.currentThread().getName());
        progressDialog = new ProgressDialog(context);
        progressDialog.setTitle("标题");
        progressDialog.setMessage("Loading...");
        progressDialog.setCancelable(false);
        progressDialog.show();
    }

    private void hideProcessDialog() {
        if (progressDialog != null && progressDialog.isShowing()) {
            progressDialog.dismiss();
        }
    }

    private void showToast(String message, Context context) {
        Toast.makeText(context, message, Toast.LENGTH_SHORT).show();
    }

    private int calculateFactorial(int n) {
        int factorial = 1;
        for (int i = 1; i <= n; i++) {
            factorial *= i;
        }
        return factorial;
    }
}
